package com.atguigu.fink.chapter01.window;

import com.atguigu.fink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2022/11/28 15:09
 */
public class Flink03_TVF_2 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> stream = env
            .fromElements(
                new WaterSensor("s1", 1000L, 10),
                new WaterSensor("s1", 2000L, 10),
                new WaterSensor("s2", 3000L, 20),
                new WaterSensor("s1", 4000L, 30),
                new WaterSensor("s1", 7000L, 40),
                new WaterSensor("s1", 9000L, 40),
                new WaterSensor("s1", 12000L, 40),
                new WaterSensor("s1", 17000L, 40),
                new WaterSensor("s1", 22000L, 40),
                new WaterSensor("s1", 24000L, 40),
                new WaterSensor("s1", 30000L, 50),
                new WaterSensor("s1", 31000L, 50)
            )
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((ws, ts) -> ws.getTs())
            
            );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts").rowtime(), $("vc"));
        tEnv.createTemporaryView("sensor", table);
        
        tEnv.sqlQuery("select " +
                          "window_start, " +
                          "window_end, " +
                          "id, " +
                          "sum(vc) vc_sum " +
                          "from table( tumble( table sensor, descriptor(ts), interval '5' second ) )" +
//                          "group by window_start, window_end, grouping sets( (id), ()"
//                          "group by window_start, window_end, rollup(id)"
                          "group by window_start, window_end, cube(id)"
            )
            .execute()
            .print();
        
        
    }
}
/*
 a,b,c,d
 (a,b,c,d) (a,b,c) (a,b) (a) () rollup 上钻
 
a,b,c cube 多维立方体  2^n
  (a,b,c)
  (a,b) (a,c) (b,c)
  (a) (b) (c)
  ()
 
 */